import LocateIp2.stat0RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object LogAnalyse {
  var stat0RDD : RDD[(String,Int)] = null

  var stat1RDD : RDD[(String,String)] = null
  def main(args: Array[String]): Unit = {
    // 1、创建SparkContext
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    val N = 3

    // 2.1、计算独立IP数
    val logRecords: RDD[String] = sc.textFile("file:///D:\\BaiduNetdiskDownload\\lagoubigdata\\fourthPhrase\\大数据正式班第四阶段模块一\\scala编程\\0-讲义和代码\\代码\\sparkPartOne\\data\\cdn.txt")

    println("===========计算独立IP数=============")
    stat0RDD = logRecords.map {line =>
      val fields :Array[String] = line.split(" ")
      //var index = fields(0).lastIndexOf(".")
      //fields(0) = fields(0).substring(0,index)
      (fields(0),1)
    }
    stat0RDD.reduceByKey(_+_).collect().sortBy(_._1).foreach(println);


    //2.2、统计每个视频独立IP数（视频的标志：在日志文件的某些可以找到 *.mp4，代表一个视频文件）
    println("==========统计每个视频独立IP数（视频的标志：在日志文件的某些可以找到 *.mp4，代表一个视频文件）============")
    stat1RDD = logRecords.map {line =>
      val fields :Array[String] = line.split(" ")
      //var index = fields(0).lastIndexOf(".")
      //fields(0) = fields(0).substring(0,index)
      (fields(0),fields(5))
    }.filter(map=>map._2.lastIndexOf(".mp4")>0).map(a=>(a._2,a._1))
    var newstat1Rdd : RDD[(String,String)] = stat1RDD.distinct()
    //var newstat1Rdd : RDD[(String,String)] = sc.makeRDD(stat1RDD.distinct().collect())
    newstat1Rdd.map(a=>(a._1,1)).reduceByKey(_+_).collect().sortBy(_._1).foreach(println);

    // 2、统计每一个省份每一个小时的TOP3广告ID
    println("统计每一个省份每一个小时的TOP3广告ID")
    logRecords.map { line =>
      val fields: Array[String] = line.split("\\s+")
      var startIndex = fields(3).indexOf(":")
      var hour = fields(3).substring(startIndex+1,startIndex+3)
      (hour, 1)
    }.reduceByKey(_+_)
      .collect.foreach(println)

  }

}
